src/sentry/remote_subscriptions/consumers/queue_consumer.py (4)

49-54: Minor: _get_partition_lock has a benign TOCTOU window.

Two threads could both see None from .get() and create separate Lock objects, but setdefault is atomic under CPython's GIL, so only one lock wins and both callers get the same instance. This is safe in practice, but if you want to be explicit, you could use a single defaultdict(threading.Lock) for partition_locks or protect with a global lock.

---

83-93: get_committable_offsets is O(offset_range), not O(tracked_offsets).

The range(start, max_offset + 1) loop iterates over every integer in the offset range, not just tracked offsets. If a partition has a sparse set of tracked offsets (e.g., offsets 100 and 10000), this iterates ~9900 times even though only 2 offsets are tracked. In practice this is bounded by processing throughput, but worth keeping in mind if offset values become sparse.

---

273-291: Commit loop waits 1 second before first commit check.

The self.shutdown_event.wait(1.0) at the top of the loop body (line 276) means the first commit check doesn't happen until ~1 second after the strategy is created, regardless of how quickly messages are processed. This is likely fine for production (matches ONCE_PER_SECOND policy), but it explains why some tests need to wait longer for commits to appear.

---

293-325: Exception during submit after add_offset but before queue_pool.submit could orphan offsets.

If self.decoder succeeds, self.grouping_fn succeeds, but self.queue_pool.submit fails (line 315), the exception handler (lines 317-325) checks isinstance(message.value, BrokerValue) and adds+completes the offset. However, queue_pool.submit calls offset_tracker.add_offset before work_queue.put (line 211-212). If put raises, the offset is added but never completed, and the exception handler adds it again. The add_offset call at line 211 adds the offset to all_offsets and outstanding, then the handler at line 320-325 adds it again (it's a set, so no duplicate), and completes it. This is actually safe because sets handle duplicates, but the flow is subtle and could benefit from a comment.

src/sentry/remote_subscriptions/consumers/result_consumer.py (1)

205-212: Consider using elif for the mode dispatch chain.

The mode flags are mutually exclusive, but the current if/if/if/else structure doesn't make that obvious. Using elif would make the intent clearer and prevent accidental fall-through if a future change removes an early return.

Proposed refactor

     def create_with_partitions(
         self,
         commit: Commit,
         partitions: Mapping[Partition, int],
     ) -> ProcessingStrategy[KafkaPayload]:
         if self.batched_parallel:
             return self.create_thread_parallel_worker(commit)
-        if self.parallel:
+        elif self.parallel:
             return self.create_multiprocess_worker(commit)
-        if self.thread_queue_parallel:
+        elif self.thread_queue_parallel:
             return self.create_thread_queue_parallel_worker(commit)
         else:
             return self.create_serial_worker(commit)

tests/sentry/uptime/consumers/test_results_consumer.py (1)

1824-1872: Polling loop could time out without asserting failure.

Lines 1860-1867: if the loop times out (queue never empties or no commits appear), execution falls through to the assertions at lines 1869-1870, which would fail with a less informative error. Consider adding an explicit timeout assertion like the other tests do: assert ... , "Processing did not complete in time".

Proposed improvement

             queue_pool = factory.queue_pool
             max_wait = 20
+            completed = False
             for _ in range(max_wait):
                 assert queue_pool is not None
                 stats = queue_pool.get_stats()
                 if stats["total_items"] == 0 and len(committed_offsets) > 0:
+                    completed = True
                     break
-
                 time.sleep(0.1)
+            assert completed, "Processing or commit did not complete in time"